﻿//===============================================================================
//	SyncQueue.h
//	同步队列
//	保存指定类型的消息，存取加锁
//===============================================================================

#pragma once

#include <deque>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <iostream>
#include <cstdint>
#include "user_define.h"
#include "log.h"

template<typename T>
class SyncQueue
{
public:
	//---------------------------------------------------------------------------
	SyncQueue(uint32_t maxSize)
		: m_maxSize(maxSize)
		, m_needStop(false)
	{
		_safe_guard_begin;
		_safe_guard_end;
	}

	//---------------------------------------------------------------------------
	~SyncQueue(void)
	{}

public:
	//---------------------------------------------------------------------------
	void Put(const T& x)
	{
		_safe_guard_begin;

		add(x);
		
		_safe_guard_end;
	}

	//---------------------------------------------------------------------------
	void Put(T&& x)
	{
		_safe_guard_begin;

		add(std::forward<T>(x));

		_safe_guard_end;
	}

	//---------------------------------------------------------------------------
	void Take(std::deque<T>& list)
	{
		_safe_guard_begin;

		if (m_needStop)
		{
			return;
		}

		if (IsEmpty())
		{
			return;
		}

		//	一次加锁，获取全部数据，以提高性能
		std::unique_lock<std::recursive_mutex> locker(m_mutex);
		//m_notEmpty.wait(locker, [this] {return m_needStop || notEmpty(); });

		list = std::move(m_queue);
		//m_notFull.notify_one();

		_safe_guard_end;
	}

	//---------------------------------------------------------------------------
	void Take(T& t)
	{
		_safe_guard_begin;

		if (m_needStop)
		{
			return;
		}

		if (IsEmpty())
		{
			return;
		}

		std::unique_lock<std::recursive_mutex> locker(m_mutex);
		//	不满足条件时，释放mutex，并将线程置于waiting状态。
		//	等待其他线程调用notify_one/notify_all，将其唤醒。
		//	处于waiting状态的线程被唤醒时，条件变量先获取mutex，
		//	然后检查条件是否满足，如果满足，继续执行，
		//	否则释放mutex，继续等待。
		//m_notEmpty.wait(locker, [this] {return m_needStop || notEmpty(); });

		t = m_queue.front();
		m_queue.pop_front();

		//m_notFull.notify_one();
		_safe_guard_end;
	}

	//---------------------------------------------------------------------------
	void Stop(void)
	{
		_safe_guard_begin;
		
		{
			std::lock_guard<std::recursive_mutex> locker(m_mutex);
			m_needStop = true;
			m_queue.clear();
		}

		//	notify_one/notify_all会唤醒一个在等待的线程，
		//	线程唤醒后先获取mutex再检查条件是否满足，
		//	如果此时mutex被保护，则需要析构释放才能获取，
		//	在lock_guard外调用，性能会有所提高。
		//m_notEmpty.notify_all();
		//m_notFull.notify_all();

		_safe_guard_end; 
	}

	//---------------------------------------------------------------------------
	bool IsEmpty(void)
	{
		_safe_guard_begin;

		std::lock_guard<std::recursive_mutex> locker(m_mutex);
		return (m_queue.empty());

		_safe_guard_end;

		return false;
	}

	//---------------------------------------------------------------------------
	bool IsFull(void)
	{
		_safe_guard_begin;

		std::lock_guard<std::recursive_mutex> locker(m_mutex);
		return (m_queue.size() >= m_maxSize);

		_safe_guard_end;

		return false;
	}

	//---------------------------------------------------------------------------
	uint32_t Size(void)
	{
		_safe_guard_begin;

		std::lock_guard<std::recursive_mutex> locker(m_mutex);
		return ((uint32_t)(m_queue.size()));

		_safe_guard_end;

		return 0;
	}

	//---------------------------------------------------------------------------
	uint32_t Count(void)
	{
		_safe_guard_begin;

		return (m_queue.size());

		_safe_guard_end;

		return 0;
	}

private:
	//---------------------------------------------------------------------------
	bool notFull(void) const
	{
		_safe_guard_begin;

		bool full = (m_queue.size() >= m_maxSize);
		if (full)
		{
			std::cout << "缓冲区满,需要等待..." << std::endl;
		}

		return (!full);

		_safe_guard_end;

		return false;
	}

	//---------------------------------------------------------------------------
	bool notEmpty(void) const
	{
		_safe_guard_begin;

		bool empty = m_queue.empty();
		if (empty)
		{
			//std::cout << "缓冲区空,需要等待...\t异步线程ID::" << std::this_thread::get_id() << std::endl;
		}

		return (!empty);

		_safe_guard_end;

		return false;
	}

	//---------------------------------------------------------------------------
	template<typename C>
	void add(C&& x)
	{
		_safe_guard_begin;

		std::unique_lock<std::recursive_mutex> locker(m_mutex);
		//m_notFull.wait(locker, [this] {return m_needStop || notFull(); });
		if (m_needStop)
		{
			return;
		}

		m_queue.push_back(std::forward<C>(x));
		//m_notEmpty.notify_one();

		_safe_guard_end;
	}

private:
	std::deque<T> m_queue;					//	缓冲区
	std::recursive_mutex m_mutex;						//	互斥量与条件变量那个结合使用
	//std::condition_variable m_notEmpty;		//	非空条件变量
	//std::condition_variable m_notFull;		//	未满条件变量
	uint32_t m_maxSize;						//	同步队列最大Size
	bool m_needStop;						//	停止标志
};

